package com.settlement.system.service.impl;

import com.alibaba.fastjson2.JSONObject;
import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
import com.settlement.system.common.constant.KafkaConstants;
import com.settlement.system.common.constant.SystemConstants;
import com.settlement.system.common.enums.PlatFormEnum;
import com.settlement.system.common.util.DateUtils;
import com.settlement.system.common.util.NumberUtils;
import com.settlement.system.common.util.StringUtils;
import com.settlement.system.handler.disruptor.MessageVo;
import com.settlement.system.mapper.SysTmBillDetailMapper;
import com.settlement.system.mapper.TmBillMapper;
import com.settlement.system.model.entity.*;
import com.settlement.system.model.form.TmBillForm;
import com.settlement.system.model.query.TmBillQuery;
import com.settlement.system.model.vo.TmBillImportVo;
import com.settlement.system.model.vo.tianmao.SysTmBillVo;
import com.settlement.system.service.*;
import lombok.RequiredArgsConstructor;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.BeanUtils;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.kafka.support.SendResult;
import org.springframework.stereotype.Service;
import org.springframework.util.Assert;

import java.time.LocalDate;
import java.time.format.DateTimeFormatter;
import java.util.*;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;

/**
 * <p>
 * 天猫账单 服务实现类
 * </p>
 *
 * @author huangrensen
 * @since 2024-04-15
 */
@Service
@RequiredArgsConstructor
public class TmBillServiceImpl extends ServiceImpl<TmBillMapper, TmBill> implements TmBillService {

    private final Logger logger = LoggerFactory.getLogger(TmBillServiceImpl.class);
    private final SysTmBillDetailMapper sysTmBillDetailMapper;

    private final SysStoreService sysStoreService;

    private final RedisTemplate redisTemplate;

    private final KafkaTemplate<String, String> kafkaTemplate;

    private final SysImportErrorDetailService importErrorDetailService;

    private final KafkaSendErrorRecordService kafkaSendErrorRecordService;
    private final SnowflakeIdWorker snowflakeIdWorker = SnowflakeIdWorker.getInstance();

    @Override
    public Page<SysTmBillVo> getPage(TmBillQuery queryParams) {
        // 查询参数
        int pageNum = queryParams.getPageNum();
        int pageSize = queryParams.getPageSize();
        // 拼接月份列表
        List<String> monthList = new ArrayList<>();
        String endMonthStr = queryParams.getEndMonth().replaceAll("-", "");
        LocalDate endMonth = LocalDate.parse(endMonthStr + "01", DateTimeFormatter.ofPattern("yyyyMMdd"));
        LocalDate startTime = LocalDate.parse(queryParams.getStartMonth().replaceAll("-", "") + "01", DateTimeFormatter.ofPattern("yyyyMMdd"));
        // 获取已经生成的表名
        Object object = redisTemplate.opsForValue().get(SystemConstants.TM_BILL);
        List<String> tableNameList;
        if (object != null) {
            tableNameList = (List<String>) object;
        } else {
            tableNameList = this.baseMapper.getAllTable();
            // 添加缓存
            redisTemplate.opsForValue().set(SystemConstants.TM_BILL, tableNameList, 1, TimeUnit.HOURS);
        }
        // 循环追加月份
        while (!startTime.isAfter(endMonth)) {
            monthList.add(startTime.format(DateTimeFormatter.ofPattern("yyyyMM")));
            startTime = startTime.plusMonths(1);
        }
        // 根据月份判断数据库表是否存在
        List<String> existTableList = new ArrayList<>();
        String tableName;
        for (String month : monthList) {
            tableName = "tm_bill_%s".formatted(month);
            if (tableNameList.contains(tableName)) {
                existTableList.add(tableName);
            }
        }
        queryParams.setTableNameList(existTableList);
        // 查询数据
        Page<TmBill> page = new Page<>(pageNum, pageSize);
        Page<SysTmBillVo> voPage = new Page<>(pageNum, pageSize);
        if (existTableList.size() == 0) {
            return voPage;
        }
        Page<TmBill> billPage = this.baseMapper.getPage(page, queryParams);
        BeanUtils.copyProperties(billPage, voPage, "records");
        List<SysTmBillVo> voList = new ArrayList<>();
        SysTmBillVo vo;
        List<String> ids = billPage.getRecords().stream().map(TmBill::getId).collect(Collectors.toList());
        SysTmBillDetail sysTmBillDetail = new SysTmBillDetail();
        List<SysTmBillDetail> allDetailList = new ArrayList<>();
        List<SysTmBillDetail> subDetailList;
        // 根据账单表名查询明细表
        for (String billTableName : existTableList) {
            billTableName = "tm_bill_detail_%s".formatted(billTableName.replace("tm_bill_", ""));
            sysTmBillDetail.setTableName(billTableName);
            sysTmBillDetail.setOrderIds(ids);
            if (!StringUtils.hasText(sysTmBillDetailMapper.existTable(billTableName))) {
                continue;
            }
            subDetailList = sysTmBillDetailMapper.getList(sysTmBillDetail);
            if (subDetailList.size() > 0) {
                allDetailList.addAll(subDetailList);
            }
        }
        SysTmBillDetail billDetail;
        SysStore sysStore;
        for (TmBill record : billPage.getRecords()) {
            vo = new SysTmBillVo();
            sysStore = sysStoreService.getById(record.getStoreId());
            if (sysStore != null) {
                vo.setStoreName(sysStore.getCommercePlatformStoreName());
            }
            BeanUtils.copyProperties(record, vo, "orderCreateTime", "orderPayTime", "sendTime", "confirmReceiveTime");
            vo.setOrderCreateTime(record.getOrderCreateTime());
            vo.setOrderPayTime(record.getOrderPayTime());
            vo.setSendTime(record.getSendTime());
            vo.setConfirmReceiveTime(record.getConfirmReceiveTime());
            billDetail = allDetailList.stream().filter(item -> item.getOrderId().equals(record.getId())).findFirst().orElse(null);
            if (billDetail != null) {
                // 填充明细
                BeanUtils.copyProperties(billDetail, vo);
            }
            voList.add(vo);
        }
        voPage.setRecords(voList);
        return voPage;
    }

    @Override
    public boolean saveTmBill(TmBillForm sysTmBillForm) {
        TmBill sysTmBill = new TmBill();
        BeanUtils.copyProperties(sysTmBillForm, sysTmBill);
        sysTmBill.setCreateTime(null);
        return this.saveOrUpdate(sysTmBill);
    }

    @Override
    public TmBillForm getTmBillFormData(Long id) {
        TmBill sysTmBill = this.getById(id);
        TmBillForm sysTmBillForm = new TmBillForm();
        BeanUtils.copyProperties(sysTmBill, sysTmBillForm);
        return sysTmBillForm;
    }

    @Override
    public boolean deleteTmBill(String ids) {
        Assert.isTrue(StringUtils.hasText(ids), "删除的数据为空");
        List<Long> idArray = Arrays.stream(ids.split(",")).map(Long::parseLong).collect(Collectors.toList());
        return this.removeByIds(idArray);
    }

    @Override
    public void importData(List<TmBillImportVo> voList) {
        List<String> billMonthList = voList.stream().map(TmBillImportVo::getBillMonth).distinct().toList();
        TmBill sysTmBill;
        for (String billMonth : billMonthList) {
            if (!StringUtils.hasText(this.baseMapper.existTable("tm_bill_%s".formatted(billMonth)))) {
                this.baseMapper.createTable("tm_bill_%s".formatted(billMonth));
            }
            // 检查是否存在明细表
            if (!StringUtils.hasText(sysTmBillDetailMapper.existTable("tm_bill_detail_%s".formatted(billMonth)))) {
                sysTmBillDetailMapper.createTable("tm_bill_detail_%s".formatted(billMonth));
            }
        }
        redisTemplate.delete(SystemConstants.TM_BILL);
        List<SysTmBillDetail> sysTmBillDetailList = new ArrayList<>();
        SysTmBillDetail billDetail;
        List<TmBill> billList = new ArrayList<>();
        for (TmBillImportVo importVo : voList) {
            sysTmBill = new TmBill();
            BeanUtils.copyProperties(importVo, sysTmBill, "orderCreateTime", "orderPayTime", "sendTime", "confirmReceiveTime");
            // ID
            sysTmBill.setId(snowflakeIdWorker.nextId(importVo.getBillMonth()));
            sysTmBill.setBillMonth(importVo.getBillMonth());
            // 店铺
            sysTmBill.setStoreId(importVo.getStoreId());
            // 订单创建时间
            sysTmBill.setOrderCreateTime(DateUtils.convertStringToLocalDateTime(importVo.getOrderCreateTime()));
            // 订单付款时间
            sysTmBill.setOrderPayTime(DateUtils.convertStringToLocalDateTime(importVo.getOrderPayTime()));
            // 发货时间
            sysTmBill.setSendTime(DateUtils.convertStringToLocalDateTime(importVo.getSendTime()));
            // 确认收货时间
            sysTmBill.setConfirmReceiveTime(DateUtils.convertStringToLocalDateTime(importVo.getConfirmReceiveTime()));
            // 保存 通过订单创建时间字段却确定保存到对应表
            sysTmBill.setTableName("tm_bill_%s".formatted(importVo.getBillMonth()));
            sysTmBill.cutString();
            this.baseMapper.insertData(sysTmBill);
            billList.add(sysTmBill);
            // 保存明细
            billDetail = new SysTmBillDetail();
            // 需要过滤的字段列表
            List<String> filterFieldList = Arrays.asList("buyerPayableAmount", "buyerPayablePostage", "buyerPayableTotalAmount", "buyerPayableActAmount", "sellerServiceAmount", "buyerServiceAmount", "purchaseRechargeAmount", "purchaseRechargeQyAmount", "purchaseConsumeAmount", "purchaseConsumeQyAmount", "purchaseConsumeAmount", "purchaseConsumeQyAmount", "refundAmount", "cardDeduction", "jfbDeduction", "payoutsAmount", "personalRedPackets", "cashbackAmount", "delaySendIndemnityAmount", "tradePayAmount", "afterSalesAmount", "hbPayableAmount", "creditCardPayableAmount", "popularizeAmount", "jhsAmount", "depositAmount", "productQty");
            BeanUtils.copyProperties(importVo, billDetail, filterFieldList.toArray(new String[0]));
            billDetail.setOrderId(sysTmBill.getId());
            // 买家应付货款
            billDetail.setBuyerPayableAmount(NumberUtils.convert(importVo.getBuyerPayableAmount()));
            // 买家应付邮费
            billDetail.setBuyerPayablePostage(NumberUtils.convert(importVo.getBuyerPayablePostage()));
            // 总金额
            billDetail.setBuyerPayableTotalAmount(NumberUtils.convert(importVo.getBuyerPayableTotalAmount()));
            // 买家实际支付金额
            billDetail.setBuyerPayableActAmount(NumberUtils.convert(importVo.getBuyerPayableActAmount()));
            // 卖家服务费
            billDetail.setSellerServiceAmount(NumberUtils.convert(importVo.getSellerServiceAmount()));
            // 买家服务费
            billDetail.setBuyerServiceAmount(NumberUtils.convert(importVo.getBuyerServiceAmount()));
            // 购物金充值本金
            billDetail.setPurchaseRechargeAmount(NumberUtils.convert(importVo.getPurchaseRechargeAmount()));
            // 购物金充值权益金
            billDetail.setPurchaseRechargeQyAmount(NumberUtils.convert(importVo.getPurchaseRechargeQyAmount()));
            // 购物金消费本金
            billDetail.setPurchaseConsumeAmount(NumberUtils.convert(importVo.getPurchaseConsumeAmount()));
            // 购物金消费权益金
            billDetail.setPurchaseConsumeQyAmount(NumberUtils.convert(importVo.getPurchaseConsumeQyAmount()));
            // 退款金额
            billDetail.setRefundAmount(NumberUtils.convert(importVo.getRefundAmount()));
            // 天猫卡券抵扣
            billDetail.setCardDeduction(NumberUtils.convert(importVo.getCardDeduction()));
            // 集分宝抵扣
            billDetail.setJfbDeduction(NumberUtils.convert(importVo.getJfbDeduction()));
            // 打款商家金额
            billDetail.setPayoutsAmount(NumberUtils.convert(importVo.getPayoutsAmount()));
            // 含应开票给个人的个人红包
            billDetail.setPersonalRedPackets(NumberUtils.convert(importVo.getPersonalRedPackets()));
            // 返现金额
            billDetail.setCashbackAmount(NumberUtils.convert(importVo.getCashbackAmount()));
            // 延迟发货主动赔付金额
            billDetail.setDelaySendIndemnityAmount(NumberUtils.convert(importVo.getDelaySendIndemnityAmount()));
            // 交易付款
            billDetail.setTradePayAmount(NumberUtils.convert(importVo.getTradePayAmount()));
            // 售后退款
            billDetail.setAfterSalesAmount(NumberUtils.convert(importVo.getAfterSalesAmount()));
            // 花呗支付服务费
            billDetail.setHbPayableAmount(NumberUtils.convert(importVo.getHbPayableAmount()));
            // 信用卡支付服务费
            billDetail.setCreditCardPayableAmount(NumberUtils.convert(importVo.getCreditCardPayableAmount()));
            // 淘宝内容推广服务费
            billDetail.setPopularizeAmount(NumberUtils.convert(importVo.getPopularizeAmount()));
            // 聚划算佣金
            billDetail.setJhsAmount(NumberUtils.convert(importVo.getJhsAmount()));
            // 保证金退款
            billDetail.setDepositAmount(NumberUtils.convert(importVo.getDepositAmount()));
            // 宝贝总数量
            billDetail.setProductQty(NumberUtils.convertInteger(importVo.getProductQty()));
            // 保存到对应表
            billDetail.setTableName("tm_bill_detail_%s".formatted(sysTmBill.getBillMonth()));
            // 主订单编号
            billDetail.setTmMainOrderNo(importVo.getMainOrderNo());
            billDetail.cutString();
            sysTmBillDetailList.add(billDetail);
        }
        if (sysTmBillDetailList.size() > 0) {
            sysTmBillDetailList.forEach(sysTmBillDetailMapper::insertData);
        }
        // 删除缓存
        redisTemplate.delete(SystemConstants.TM_BILL);
        // 发送队列消息通知
        MessageVo vo;
        for (TmBill tmBill : billList) {
            vo = new MessageVo();
            vo.setDataId(tmBill.getId());
            vo.setStoreId(tmBill.getStoreId());
            vo.setBillMonth(tmBill.getBillMonth());
            // 根据字典中的备注查找对应的数据库表字段名
            vo.setTableNamePrefix("tm_bill_");
            // 从字段管理里找到对应的字段类型编码，获取对应的billType
            vo.setDictCode(PlatFormEnum.TMALL_CONFIG.getValue());
            // 推送到转换标准账单的消息队列
            vo.setUuid(snowflakeIdWorker.nextId());
            CompletableFuture<SendResult<String, String>> completableFuture = kafkaTemplate.send(KafkaConstants.HWG_TOPIC_BILL_QUEUE, JSONObject.toJSONString(vo));
            MessageVo finalVo = vo;
            completableFuture.whenComplete((result, throwable) -> {
                if(Objects.nonNull(throwable)){
                    throwable.printStackTrace();
                    KafkaSendErrorRecord errorRecord = new KafkaSendErrorRecord();
                    errorRecord.setDataText(JSONObject.toJSONString(finalVo));
                    errorRecord.setRemark(throwable.getMessage());
                    errorRecord.setQueueName(KafkaConstants.HWG_TOPIC_BILL_QUEUE);
                    errorRecord.setUuid(finalVo.getUuid());
                    kafkaSendErrorRecordService.checkAndSave(errorRecord);
                }
            });
        }
    }

    @KafkaListener(id = KafkaConstants.HWG_TOPIC_TM_BILL, topics = KafkaConstants.HWG_TOPIC_TM_BILL, groupId = "yzConsumerGroup")
    public void importData(List<ConsumerRecord<String, String>> records, Acknowledgment acknowledgment) {
        logger.info("【tm-bill-queue】当前消费者本次拉取的数据量为：{}", (Objects.isNull(records)? 0 : records.size()));
        if(Objects.isNull(records) || records.isEmpty()){
            logger.error("【tm-bill-queue】无消费数据...");
        }
        String orderNo = "";
        for (ConsumerRecord<String, String> record : records) {
            String importVoStr = record.value();
            try {
                TmBillImportVo importVo = JSONObject.parseObject(importVoStr, TmBillImportVo.class);
                orderNo = importVo.getOrderNo();
                this.importData(Collections.singletonList(importVo));
            } catch (Exception e) {
                logger.info("消费失败：{}", e.getMessage());
                // 保存失败数据
                SysImportErrorDetail detail = new SysImportErrorDetail();
                detail.setPlatformAndType("天猫账单");
                detail.setDataText(importVoStr);
                detail.setOrderNo(orderNo);
                detail.setQueueName(KafkaConstants.HWG_TOPIC_TM_BILL);
                detail.setRemark(e.getMessage());
                importErrorDetailService.checkAndCreate(detail);
            }
        }
        // 确认消费
        acknowledgment.acknowledge();
    }

    @Override
    public List<TmBillImportVo> findByIds(String idsStr) {
        if (!StringUtils.hasText(idsStr)) {
            return new ArrayList<>();
        }
        List<Long> ids = Arrays.stream(idsStr.split(",")).map(Long::parseLong).collect(Collectors.toList());
        List<TmBill> sysTmBillList = this.getBaseMapper().selectBatchIds(ids);
        List<TmBillImportVo> voList = new ArrayList<>();
        for (TmBill sysTmBill : sysTmBillList) {
            TmBillImportVo vo = new TmBillImportVo();
            BeanUtils.copyProperties(sysTmBill, vo);
            voList.add(vo);
        }
        return voList;
    }
}
